package com.oildatacollection.datareceive.datastore;

import com.oildatacollection.database.InfluxdbConnection;
import com.oildatacollection.message.RecvMessage;
import javafx.util.Pair;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.InfluxDBIOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

@Slf4j
public class StoreManagerInfluxImpl implements StoreManager {
    private final InfluxdbConnection influxConn;

    public StoreManagerInfluxImpl() {
        influxConn = new InfluxdbConnection("dataInsert");
    }

    @Override
    public void connectDB() {
        try {
            influxConn.influxDbBuild();
        } catch (InfluxDBIOException e) {
            log.error("接收队列--dataInsert: 数据库连接失败", e);
        }
        log.trace("接收队列--dataInsert: 数据库连接成功");
    }

    /**
     * 插入数据库
     * @param msg
     * @param curTime
     * @param timeUnit
     */
    private void insertDB(RecvMessage msg, long curTime, TimeUnit timeUnit) {
        String tableName = influxConn.getUseTableName();

        int wellID = (int) msg.getValue("wellID");
        Map<String, String> tags = new HashMap<>();
        tags.put("wellID", Integer.toString(wellID));

        Map<String, Object> fields = new HashMap<>();
        msg.resetIter();
        while (msg.hasNext()) {
            Pair<String, Object> item = msg.getNext();
            String curKey = item.getKey();
            if (!Objects.equals(curKey, "wellID")) {
                Object curVal = item.getValue();
                fields.put(curKey, curVal);
            }
        }
        // 插入数据库中
        influxConn.insert(tableName, tags, fields, curTime, timeUnit);
    }

    @Override
    public void insertMessageToDB(RecvMessage msg) {
        long curTime = System.currentTimeMillis();
        insertDB(msg, curTime, TimeUnit.MILLISECONDS);
    }
}
